大家好,鐵人賽第十九天
昨天,我們用 fast-mcp
成功地搭建了一個最簡潔的 "Hello, World!" 應用,感受了它輕量、快速的設計哲學。今天,我們將不再滿足於簡單的問候,而是要動真格的——我們要用 fast-mcp
框架,來重新打造我們在 Day 15 用 adk-mcp
實現過的、那個能串接真實 API 的「天氣查詢工具」!
這次實戰將讓我們深刻體會到,如何用 fast-mcp
來組織一個更複雜的專案,並處理環境變數、外部 API 呼叫等真實世界的開發需求。
在開始之前,我們需要為這個更複雜的專案做一些準備。
.env
)為了安全起見,我們絕不能將 API 金鑰直接寫在程式碼裡。
.env
的檔案。CWA_AUTH_TOKEN="這裡是你的中央氣象署API金鑰"
my_server.py
)現在,讓我們來編寫 fast-mcp
版本的伺服器。我們將採用一個非常專業的實踐:將核心的 API 查詢邏輯與對外暴露的 MCP 工具進行分離。
用以下程式碼覆蓋你的 my_server.py
檔案:
import os
import urllib.parse
import requests
from dotenv import load_dotenv
from fastmcp import FastMCP
# 載入 .env 檔案中的環境變數
load_dotenv()
# --- 核心天氣查詢邏輯 ---
# (這部分與我們在 ADK 專案中寫的幾乎一樣,證明了核心邏輯可以輕鬆遷移)
TAIWAN_CITIES = [
"臺北市", "新北市", "桃園市", "臺中市", "臺南市", "高雄市",
"基隆市", "新竹市", "嘉義市", "新竹縣", "苗栗縣", "彰化縣",
"南投縣", "雲林縣", "嘉義縣", "屏東縣", "宜蘭縣", "花蓮縣",
"臺東縣", "澎湖縣", "金門縣", "連江縣"
]
def fetch_weather_from_cwa(city: str) -> str:
"""從中央氣象署 API 獲取天氣資訊並格式化成字串。"""
auth_token = os.getenv("CWA_AUTH_TOKEN")
if not auth_token:
return "錯誤:伺服器未設定 CWA_AUTH_TOKEN 環境變數。"
# ( ... 中間的 API 呼叫與 JSON 解析邏輯與 Day 15 相同 ... )
try:
# ... requests.get ...
# ... data parsing ...
report = (
f"為您查詢到 '{city}' 的天氣預報:\n"
f" - 時間:{time_info['startTime']} 至 {time_info['endTime']}\n"
f" - 天氣狀況:{weather_condition}\n"
f" - 降雨機率:{rain_prob}%\n"
f" - 溫度:攝氏 {min_temp} 至 {max_temp} 度"
)
return report
except Exception as e:
return f"查詢天氣時發生了預期外的錯誤:{e}"
# --- FastMCP 伺服器與工具定義 ---
mcp = FastMCP("Weather Tool Server")
@mcp.tool
def get_weather_forecast(city: str) -> str:
"""
根據提供的台灣城市名稱,取得即時天氣預報。
例如: "臺北市", "台中市", "花蓮縣"
"""
print(f"伺服器收到請求:查詢 '{city}' 的天氣...")
# 工具函式只做一件事:呼叫內部的核心邏輯
result = fetch_weather_from_cwa(city)
print("伺服器已準備好回傳結果。")
return result
if __name__ == "__main__":
mcp.run(transport="http", port=8005)
fetch_weather_from_cwa
函式中。而 @mcp.tool
裝飾的 get_weather_forecast
函式則非常乾淨,它只負責接收請求並呼叫核心邏輯。這讓程式碼非常清晰且易於維護。os.getenv("CWA_AUTH_TOKEN")
來安全地讀取 API 金鑰。my_client.py
)客戶端的工作很單純:它只需要知道工具的名稱 (get_weather_forecast
) 和需要的參數 (city
) 即可。
用以下程式碼覆蓋你的 my_client.py
檔案:
import asyncio
from fastmcp import Client
client = Client("http://localhost:8005/mcp")
async def test_weather_tool():
"""一個非同步函式,用於測試天氣查詢工具。"""
city_to_query = "臺中市"
print(f"客戶端:準備呼叫工具 'get_weather_forecast',查詢城市:{city_to_query}")
async with client:
result = await client.call_tool(
"get_weather_forecast",
{"city": city_to_query}
)
print("\n--- 成功收到 server 回傳的結果 ---")
# fastmcp 回傳的結果在 result.content 中
print(result.content[0].text)
print("---------------------------------")
if __name__ == "__main__":
asyncio.run(test_weather_tool())
第一步:啟動伺服器
fast-mcp
的 CLI 指令:fastmcp run my_server.py:mcp --transport http --port 8005
伺服器現在已經在 8005
埠口上等待接收天氣查詢請求。
第二步:執行客戶端
python my_client.py
成果驗收
你的客戶端終端機將會印出從伺服器即時獲取並格式化好的天氣資訊!
客戶端:準備呼叫工具 'get_weather_forecast',查詢城市:臺中市
--- 成功收到 server 回傳的結果 ---
[TextContent(type='text', text="為您查詢到 '臺中市' 的天氣預報:\n - 時間:2025-09-22 18:00:00 至 2025-09-23 06:00:00\n - 天氣狀況:陰時多雲\n - 降雨機率:20%\n - 溫度:攝氏 28 至 33 度", annotations=None, meta=None)]
---------------------------------
今天,我們成功地將 fast-mcp
應用於一個真實、複雜的場景中。我們不僅僅是建立了一個工具,更學會了:
fast-mcp
專案: 透過將核心邏輯與工具定義分離,讓程式碼更清晰。.env
檔案來保護我們的 API 金鑰,這是專業開發的必備技能。fast-mcp
的實戰能力: 證明了它不僅能處理簡單的任務,更能勝任串接外部 API、處理複雜資料的實戰挑戰。我們現在同時擁有 adk-mcp
和 fast-mcp
兩個版本的實用工具。明天,我們將開始我們的專案囉
!